%%% Copyright 2010 Google Inc.
%%%
%%% Licensed under the Apache License, Version 2.0 (the "License");
%%% you may not use this file except in compliance with the License.
%%% You may obtain a copy of the License at
%%%
%%%     http://www.apache.org/licenses/LICENSE-2.0
%%%
%%% Unless required by applicable law or agreed to in writing, software
%%% distributed under the License is distributed on an "AS IS" BASIS,
%%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%% See the License for the specific language governing permissions and
%%% limitations under the License.

%%% @copyright 2010 Google Inc. All Rights Reserved.
%%% @author <jacoby@google.com> Jacoby Thwaites [http://sites/madcloud]
%%% @version {@version}
%%% @doc
%%% Fieldset server. This process represents an active state fieldset
%%% which steps from vertex to vertex on a sequencing graph, collecting
%%% and emitting fieldsets from/to operations during its lifetime.
%%%
%%% A state fieldset is linked to a process, which is started by the output
%%% of a notification operation.
%%% @see process_srv:collect/2
%%%
%%% @headerfile "../include/txm.hrl"

-module(fieldset_srv).

-behaviour(gen_server).

-include("txm.hrl").
-include("process.hrl").
-include("graph.hrl").

%% Start, stop and client functions.
-export([start_link/3, stop/1]).
-export([activate/1, collect/2]).

%% Search callback functions
-export([eval/2, goal_reached/2, successors/2]).

%% Callback functions.
-export([init/1]).
-export([handle_call/3, terminate/2]).
-export([handle_cast/2, handle_info/2, code_change/3]).

%%% ---------------------------------------------------------------------------
%%% Start and stop functions
%%% ---------------------------------------------------------------------------

%% @doc Starts the fieldset server linked to the calling process.
%% The returned PID is used to reference the fieldset server.
%% @spec start_link(ref(), term(), list()) -> {ok, pid()}
start_link(CompRef, Cookie, Fields) ->
  FieldValues = dict:from_list(Fields),
  FieldsetState = #?FIELDSET_STATE{compRef = CompRef,
                                   cookie = Cookie,
                                   fieldValues = FieldValues},
  gen_server:start_link(?MODULE, FieldsetState, []).

%% @doc Stops the fieldset server.
stop(FieldsetSrv) ->
  gen_server:call(FieldsetSrv, ?STOP,
                  ?CALL_TIMEOUT).

%%% ---------------------------------------------------------------------------
%%% Client functions
%%% ---------------------------------------------------------------------------

%% Activates the fieldset server, which starts sequencing.
activate(FieldsetSrv) ->
  gen_server:cast(FieldsetSrv, ?ACTIVATE).

%% Delivers operation output to the fieldset server, which
%% continues sequencing.
collect(FieldsetSrv, #?EVENT_FIELDSET{} = EventFieldset) ->
  gen_server:cast(FieldsetSrv, {?COLLECT, EventFieldset}).

%%% ---------------------------------------------------------------------------
%%% Callback gen_server functions
%%% ---------------------------------------------------------------------------

%% @doc Inits the fieldset_srv using the supplied state record.
%% We ensure normal exit is trapped.
%% @spec init(string() | ref()) -> {ok, #state_fieldset{}}
init(State) ->
  ?DEBUG([init]),
  process_flag(trap_exit, true),
  ?OK_RETURN(State).

%% No action on terminate.
%% @private
terminate(_Reason, _State) ->
  ?DEBUG([terminate]),
  ?OK_RETURN.

%% Upon activation, the fieldset_srv processes the first sequence.
handle_cast(?ACTIVATE, State) ->
  case sequence(State) of
    {stop, NewState} ->
      {stop, normal, NewState};
    {continue, NewState} ->
      {noreply, NewState}
  end;

%% Post activation, we handle RR operation output collection.
handle_cast({?COLLECT, EventFieldset}, State) ->

  % Handle the collection.
  case do_collection(EventFieldset, State) of
    {stop, NewState} ->
      {stop, normal, NewState};
    {continue, NewState} ->
      {noreply, NewState}
  end.

%%% ---------------------------------------------------------------------------
%%% Collection functions
%%% ---------------------------------------------------------------------------

%% Handle collection of an event fieldset output from an RR operation.
%%
%% @spec do_collection(#event_fieldset{}, #fieldset_state{}) ->
%%          {stop, #fieldset_state{}} |
%%          {continue, #fieldset_state{}}
do_collection(#?EVENT_FIELDSET{opRef = OpRef,
                               fields = Fields},
              #?FIELDSET_STATE{
               collectAgenda = CollectAgenda,
               fieldValues = FieldValues} = State) ->

  % Get the output values into our fieldset.
  NewFieldValues = lists:foldl(fun({Ref, Value}, Acc) ->
                                    dict:store(Ref, Value, Acc)
                               end, FieldValues, Fields),

  % Perform any emissions using these new values and returning the new state.
  InterimState = State#?FIELDSET_STATE{
                       collectAgenda = lists:delete(OpRef, CollectAgenda),
                       fieldValues = NewFieldValues},
  #?FIELDSET_STATE{collectAgenda = NewCollectAgenda} =
     NewState = emissions(InterimState),

  % If the collect agenda is now empty, resequence.
  case NewCollectAgenda of
    [] ->
      sequence(NewState);
    [_|_] ->
      {continue, NewState}
    end.

%%% --------------------------------------------------------------------------
%%% Sequencing functions.
%%% --------------------------------------------------------------------------

%% @doc Computes and emits all events that are possible, then either
%% returns a stop tuple, or a continue tuple with the new state.
%% @spec sequence(#fieldset_state{}) -> {stop, #fieldset_state{}} |
%%                                      {continue, #fieldset_state{}}
sequence(#?FIELDSET_STATE{compRef = CompRef,
                          fieldValues = FieldValues,
                          satisfied = Satisfied} = State) ->
  #?PROCESS_CACHE{graph = Graph,
                  vertices = Vertices} =
     cache_srv:get_cache_record(?PROCESS_CACHE, CompRef),
  Start = graph:get_vertex(Vertices,
                           sets:from_list(dict:fetch_keys(FieldValues))),
  VertexList = search:search(Start, {?MODULE, State}),

  % Compute the ops which we add to the emissions list, truncating after the
  % first RR op with multi-way output.
  case compute_emissions(Graph, VertexList, Satisfied) of
    [] ->
      % Server stops if there are no further ops to which input can be sent.
      {stop, State};
    EmitAgenda ->
      #?FIELDSET_STATE{collectAgenda = CollectAgenda} =
         NewState =
           emissions(State#?FIELDSET_STATE{emitAgenda = EmitAgenda}),

      % Server stops if emissions will result in no further collections
      % (i.e. state can't change any more), otherwise awaits collections.
      case CollectAgenda of
        [] ->
          {stop, NewState};
        [_|_] ->
          {continue, NewState}
      end
  end.

%% @doc Return the list of OO and IRR op refs satisfied by the vertices and
%% RR ops which generate the edges between them, up to and including
%% the first RR op that has multiple outputs.
%% @spec compute_emissions(term(), list(), set()) -> list()
compute_emissions(_Graph, [], _Satisfied) -> [];
compute_emissions(Graph, VertexList, Satisfied) ->
  EmitSet = compute_emissions_set(Graph, VertexList, Satisfied, sets:new()),
  sets:to_list(EmitSet).

%% Recurse over non-empty list, the base case is one element.
%% #1
compute_emissions_set(Graph, [V], Satisfied, AccSet) ->
  {V, #?VERTEX_LABEL{satisfies = Satisfies}} = digraph:vertex(Graph, V),
  lists:foldl(fun(OpRef, Acc) ->
                   case sets:is_element(OpRef, Satisfied) of
                     true ->
                       Acc;
                     false ->
                       sets:add_element(OpRef, Acc)
                   end
              end, AccSet, Satisfies);

%% Append unsatisfied IRR & OO ops referenced in the near vertex, and the RR
%% op referenced in the edge.
%% Stop if the RR op has more than 1 output, otherwise recurse starting
%% at the far vertex.
%% #2
compute_emissions_set(Graph, [V1, V2 | Rest], Satisfied, AccSet) ->

  {RR, Count} =  get_generator(Graph, V1, V2),
  NewSet =
    sets:add_element(RR, compute_emissions_set(Graph, [V1], Satisfied, AccSet)),
  case Count of
    1 ->
      compute_emissions_set(Graph, [V2 | Rest], Satisfied, NewSet);
    Count when Count > 1 ->
      NewSet
    end.

%% @doc Returns a generator op with output count that links the two
%% vertices `V1' and `V2'. There could be more than one such op, but
%% only the first found is returned.
get_generator(Graph, V1, V2) ->
  OutEdges = digraph:out_edges(Graph, V1),
  txm_util:first(fun(Edge) ->
             case digraph:edge(Graph, Edge) of
               {Edge, V1, V2, #?EDGE_LABEL{generator = OpRef}} ->
                 ?OK_RETURN(#?OP{outputs = Outputs}) =
                   prog_srv:get_record(OpRef),
                 {OpRef, nr_set_dict:count_sets(Outputs)};
               _ -> false
               end
        end, OutEdges).

%% @doc Emit all possible inputs and update state by moving emit agenda items
%% to the collect agenda where appropriate (ie for RR and IRR ops).
%% @spec emissions(#fieldset_state{}) -> #fieldset_state{}
emissions(#?FIELDSET_STATE{compRef = CompRef,
                           cookie = Cookie,
                           emitAgenda = EmitAgenda,
                           collectAgenda = CollectAgenda,
                           satisfied = Satisfied,
                           fieldValues = FieldValues} = State) ->

  % Emit input to each op in the emit agenda for which we have all the input.
  Emitted = lists:foldl(
              fun(OpRef, Acc) ->
                   ?OK_RETURN(#?OP{input = Input,
                                   opType = OpType}) =
                     prog_srv:get_record(OpRef),
                   FieldSet = sets:from_list(dict:fetch_keys(FieldValues)),
                   InputSet = nr_set:ref_set(Input),
                   case sets:is_subset(InputSet, FieldSet) of
                     true ->
                       Fields = fields_from(InputSet, FieldValues),
                       emit(OpRef, CompRef, Cookie, Fields),
                       [{OpRef, OpType} | Acc];
                     false ->
                       Acc
                     end
              end, [], EmitAgenda),

  % Compute new emit/collect agendas and the new satisfied consumers set.
  {NewEmitAgenda, NewCollectAgenda, NewSatisfied} =
    lists:foldl(fun({OpRef, OpType}, {Es, Cs, Ss}) ->
                     case OpType of
                       ?OP_TYPE_OO ->
                         {lists:delete(OpRef, Es),
                          Cs,
                          sets:add_element(OpRef, Ss)};
                       ?OP_TYPE_IRR ->
                         {lists:delete(OpRef, Es),
                          Cs,
                          sets:add_element(OpRef, Ss)};
                       ?OP_TYPE_RR ->
                         {lists:delete(OpRef, Es),
                          [OpRef | Cs],
                          Ss}
                     end
                end, {EmitAgenda, CollectAgenda, Satisfied}, Emitted),

  % Return the updated state.
  State#?FIELDSET_STATE{emitAgenda = NewEmitAgenda,
                        collectAgenda = NewCollectAgenda,
                        satisfied = NewSatisfied}.

%% @doc Makes a proplist of {ref(), term()} holding each ref() in `Set' and
%% the term() value looked up in `Dict'.
%% @spec fields_from(set(), dict()) -> proplist()
fields_from(Set, FieldValues) ->
  sets:fold(fun(FieldRef, Acc1) ->
                 Value =
                   dict:fetch(FieldRef, FieldValues),
                 [{FieldRef, Value} | Acc1]
            end,
            [], Set).

%% @doc Emits input to the operation specified by `OpRef', taking the data
%% from the supplied `Fieldset'.
%% @spec emit(ref(), ref(), term(), set()) -> ok
emit(OpRef, CompRef, Cookie, Fields) ->
  case view_db:matching_views(CompRef, Cookie, OpRef) of
    [] ->
      ?WARNING([no_view, {prog, CompRef}, {op, OpRef}, {cookie, Cookie}]);
    ViewFsms ->
      Closest = node_srv:closest(ViewFsms),
      EventFieldset = #?EVENT_FIELDSET{opRef = OpRef,
                                       setName = ?INPUT_SET_NAME,
                                       cookie = Cookie,
                                       fields = Fields},
      view_fsm:handle(Closest, self(),  EventFieldset)
    end.

%%% --------------------------------------------------------------------------
%%% Search module callback functions.
%%%   eval(SearchState, Context) -> integer()
%%%     where a lower eval means higher priority on the search agenda.
%%%
%%%   successors(SearchState, Context) -> list()
%%%     being a list of successor states.
%%%
%%%   goal_reached(SearchState, Context) -> true | false
%%%     defines whether search is complete.
%%% --------------------------------------------------------------------------

%% @doc We don't evaluate vertices right now.
eval(_Vertex, _State) -> 0.

%% doc The successors to the supplied vertex are obtained from the graph.
successors(Vertex, #?FIELDSET_STATE{compRef = CompRef}) ->
  #?PROCESS_CACHE{graph = Graph} =
    cache_srv:get_cache_record(?PROCESS_CACHE, CompRef),
  digraph:out_neighbours(Graph, Vertex).

%% @doc Goal is reached if at least one op satisfied by the vertex
% has not already been satisfied by this fieldset.
goal_reached(Vertex, #?FIELDSET_STATE{compRef = CompRef,
                                      satisfied = Satisfied}) ->
  #?PROCESS_CACHE{graph = Graph} =
     cache_srv:get_cache_record(?PROCESS_CACHE, CompRef),
  {Vertex, #?VERTEX_LABEL{satisfies = Satisfies}} =
    digraph:vertex(Graph, Vertex),
  txm_util:first(fun(OpRef) ->
             not(sets:is_element(OpRef, Satisfied))
        end, Satisfies).

%%% --------------------------------------------------------------------------
%%% Unimplemented behaviour callbacks.
%%% --------------------------------------------------------------------------
code_change(_, _, _) -> throw(?REASON_NOT_IMPLEMENTED).
handle_call(_, _, _) -> throw(?REASON_NOT_IMPLEMENTED).
handle_info(_, _) -> throw(?REASON_NOT_IMPLEMENTED).
